package com.shujia.flink.state;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Demo4Fraud {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
         *001,100
         *001,200
         *001,300
         *001,0.1
         *001,600
         *001,100
         *001,0.4
         *001,200
         *001,600
         */
        //创建kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("master:9092,node1:9092,node2:9092")//broker列表
                .setTopics("event")//指定topic
                .setGroupId("Demo5KafkaSource")//指定消费者组，保证一条数据在一个组内只消费一次
                .setStartingOffsets(OffsetsInitializer.latest())//指定起始消费的位置
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        //使用kafka source 创建流（无界流）
        DataStream<String> kafkaDS = env
                .fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        //解析数据
        DataStream<Transaction> transactions = kafkaDS.map(line -> {
            String[] split = line.split(",");
            String id = split[0];
            Double price = Double.parseDouble(split[1]);
            return new Transaction(id, price);
        });

        //对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
        KeyedStream<Transaction, String> keyByDS = transactions.keyBy(Transaction::getId);

        //process:相当于flatMap，比flatMap多个时间信息
        DataStream<Alert> alertsDS = keyByDS
                .process(new KeyedProcessFunction<String, Transaction, Alert>() {
                    ValueState<Boolean> flagState;
                    ValueState<Double> minPriceState;
                    ValueState<Long> tsState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        RuntimeContext runtimeContext = getRuntimeContext();
                        //初始化保存标记的状态
                        flagState = runtimeContext
                                .getState(new ValueStateDescriptor<>("flag", Types.BOOLEAN));

                        //保存小于1的交易金额
                        minPriceState = runtimeContext
                                .getState(new ValueStateDescriptor<>("minPrice", Types.DOUBLE));


                        tsState = runtimeContext
                                .getState(new ValueStateDescriptor<>("ts", Types.LONG));
                    }

                    //processElement:每一条数据执行一次
                    @Override
                    public void processElement(Transaction trans,
                                               KeyedProcessFunction<String, Transaction, Alert>.Context ctx,
                                               Collector<Alert> out) throws Exception {
                        //获取key
                        String id = ctx.getCurrentKey();

                        if (flagState.value() != null && flagState.value()) {
                            //紧跟着一个大于 $500 的交易
                            //两次交易需要在10内完成
                            if (ctx.timestamp() - tsState.value() < 10000 && trans.getPrice() > 500) {
                                System.out.println("紧跟着一个大于 $500 的交易");
                                //输出一个报警信息
                                Alert alert = new Alert(id, tsState.value(), minPriceState.value(), ctx.timestamp(), trans.getPrice());
                                //将数据发送到下游
                                out.collect(alert);
                            }
                            //清空状态
                            flagState.clear();
                            minPriceState.clear();
                            tsState.clear();
                        }
                        //出现小于 $1 美元的交易
                        if (trans.getPrice() < 1) {
                            System.out.println("出现小于 $1 美元的交易");
                            flagState.update(true);
                            //保存小于1的金额
                            minPriceState.update(trans.getPrice());
                            //时间戳保存到状态中
                            tsState.update(ctx.timestamp());
                        }
                    }
                });

        alertsDS.print();

        env.execute();
    }
}

@Data
@AllArgsConstructor
class Transaction {
    private String id;
    private Double price;
}

@Data
@AllArgsConstructor
class Alert {
    private String id;
    private Long minTs;
    private Double minPrice;
    private Long maxTs;
    private Double maxPrice;
}